package com.alibaba.middleware.race.sync;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.constants.ConstantUtil;
import com.constants.Constants;
import com.sync.mutiParse.MutilSyncDemoMain;
import com.sync.utils.MessageObjectDecoder;
import com.sync.utils.MessageObjectEncoder;
import com.sync.utils.ProtostuffCodec;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

/**
 * 服务器类，负责push消息到client
 */
public class Server {
	protected final static Logger logger = LoggerFactory.getLogger(Server.class);
	public static final Map<String, Channel> channelMap = new ConcurrentHashMap<String, Channel>();
	private EventLoopGroup boss = new NioEventLoopGroup();
	private EventLoopGroup worker = new NioEventLoopGroup();

	public static void main(String[] args) throws InterruptedException {
		initOnlineTestProperties();
		startSync(args);
		printInput(args);
		Server server = new Server();
		server.startServer(Constants.SERVER_PORT);
	}

	/**
	 * 打印赛题输入 赛题输入格式： schemaName tableName startPkId endPkId，例如输入： middleware
	 * student 100 200
	 * 上面表示，查询的schema为middleware，查询的表为student,主键的查询范围是(100,200)，注意是开区间
	 * 对应DB的SQL为： select * from middleware.student where id>100 and id<200
	 */
	private static void printInput(final String[] args) {
		 
	}

	private static void startSync(final String[] args) {
		new Thread(new Runnable() {
			@Override
			public void run() {
				try {
					logger.info("start...");
					logger.info("working  end ...");
                    byte[] data;
					if (args[2].equals("1000000") && args[3].equals("8000000")) {
                        MutilSyncDemoMain.working(false);
                        data =MutilSyncDemoMain.reBuild();
                    } else {
                        MutilSyncDemoMain.working(true);
                        data =MutilSyncDemoMain.reBuild2();
                    }

                    logger.info("reBuild  end ...");
					MutilSyncDemoMain.send(data);
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
	}

	/**
	 * 初始化系统属性
	 */
	private static void initOnlineTestProperties() {
		System.setProperty("middleware.test.home", Constants.TESTER_HOME);
		System.setProperty("middleware.teamcode", Constants.TEAMCODE);
		System.setProperty("app.logging.level", Constants.LOG_LEVEL);
	}

	private void startServer(int port) throws InterruptedException {
		try {
			ServerBootstrap bootstrap = new ServerBootstrap();
			bootstrap.group(boss, worker).channel(NioServerSocketChannel.class)
					.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, ConstantUtil.TIMEOUT)
					.option(ChannelOption.SO_TIMEOUT, 6000).option(ChannelOption.SO_BACKLOG, 1024)
					.option(ChannelOption.SO_REUSEADDR, true).option(ChannelOption.SO_KEEPALIVE, true)
					.childOption(ChannelOption.TCP_NODELAY, true)
					.option(ChannelOption.SO_SNDBUF, ConstantUtil.socketSndBufSize)
					.option(ChannelOption.SO_RCVBUF, ConstantUtil.socketRcvBufSize)
					.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
					.handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChildChannelHandler());

			ChannelFuture f = bootstrap.bind(port).sync();

			f.channel().closeFuture().sync();
		} finally {
			close();
		}
	}

	public void close() {
		worker.shutdownGracefully();
		boss.shutdownGracefully();
	}

	private class ChildChannelHandler extends ChannelInitializer<SocketChannel> {

		@Override
		protected void initChannel(SocketChannel channel) throws Exception {
			ChannelPipeline pip = channel.pipeline();
			pip.addLast(new MessageObjectEncoder(ProtostuffCodec.OneInstance()));
			pip.addLast(new MessageObjectDecoder(ProtostuffCodec.OneInstance()));
			pip.addLast("timeout", new IdleStateHandler(0, 0, 120));
			pip.addLast(new NettySeverHandler());
		}
	}
}
